/*
 * ===================================================================
 *  
 *      Filename: log_srv_main_framer.c
 *
 *   Description: This is the main entry for the log srv.It's a infrastructure,
 *                It does nothing except listen in the exit signal.All of the 
 *                actual jobs are done by the plugins.
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *   
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <signal.h>
#include <linux/unistd.h>
#include <unistd.h>
#include "lib/list.h"
#include "lib/util.h"

#define  __USE_GNU
#include <sched.h>
#include <pthread.h>

//#define DEBUG_LOG
/*define the array for the plugin*/
#define MAX_PLUGIN_NUM       8
pthread_rwlock_t plugin_rwlock;
struct list_head plugin_list;

pthread_mutex_t counter_mutex;
pthread_t pid_counter;
long recv_counter =0;
long send_counter =0;
long recv_traffic =0;
long color_send_counter =0;

/*****************************************************************/
int enqueue(thread_info_t *pthread,void *data)
{
    int filled_buff = 0;

    filled_buff = pthread->enqueue_counter - pthread->dequeue_counter; 

    if(filled_buff < pthread->max_queue)
    {
        if((pthread->max_queue - filled_buff) < HIGH_THRESHOLD_TRIGGER )/*free buffer counter*/
        {
            pthread->log_list[pthread->enqueue_counter&pthread->max_queue_mask] = data;
            pthread->enqueue_counter++;
            return HIGH_THRESHOLD;
        }
        else
        {
            pthread->log_list[pthread->enqueue_counter&pthread->max_queue_mask] = data;
            pthread->enqueue_counter++;
            return NORMAL_THRESHOLD;
        }
    }
    else
    {
        return URGENCY_THRESHOLD;
    }
}

void* dequeue(thread_info_t *pthread)
{
    void* result = NULL;

    if(pthread->enqueue_counter != pthread->dequeue_counter)
    {
        result = pthread->log_list[pthread->dequeue_counter&pthread->max_queue_mask];
        pthread->log_list[pthread->dequeue_counter&pthread->max_queue_mask] = NULL;
        pthread->dequeue_counter++;
    }
    else
    {
	/*pending the thread*/
	
    }

    return result;
}

void init_queue(thread_info_t *pthread, int max_queue)
{
    pthread->max_queue       = max_queue;
    pthread->max_queue_mask  = max_queue-1;
    pthread->enqueue_counter = 0;
    pthread->dequeue_counter = 0;
    pthread->log_list = (void **)malloc((size_t)((sizeof(void *)) * max_queue));
    memset(pthread->log_list, 0, sizeof(void *) * max_queue);
}

void finit_queue(thread_info_t *pthread)
{ 
    char *buff;

    /*free the buff*/
    while((buff = (char*)dequeue(pthread)))
    {
        free(buff);
    }
    /*free the buff point*/
    free((char*)pthread->log_list);
}

int assign_index_for_plugin(pc_plugin_ops *ops)
{
    ops->identifier = 0;
    if(!strcmp(ops->plugin_name(),"file target"))
    {
        ops->identifier = 1<<0;
    }
    else if(!strcmp(ops->plugin_name(),"monitor target"))
    {
        ops->identifier =  1<<1;
    }
    else if(!strcmp(ops->plugin_name(),"color target"))
    {
        ops->identifier = 1<<2;
    }
    else
    {
        ops->identifier = 1<<3;
    }

	return 0;
}
static int plugin_num = 0;
/*register a new plugin to the list*/
int mc_register_plugin(pc_plugin_ops *ops)
{
    int iret = 0;
    pthread_rwlock_wrlock(&plugin_rwlock);
    if(++plugin_num > MAX_PLUGIN_NUM)
    {
        plugin_num--;
    }
    else
    {
        list_add(&ops->list,&plugin_list);
        assign_index_for_plugin(ops);
        iret = 1;
    }
    pthread_rwlock_unlock(&plugin_rwlock);
    return iret;
}

/*delete a specific plugin from the list*/
int mc_unregister_plugin(pc_plugin_ops *ops)
{
    pthread_rwlock_wrlock(&plugin_rwlock);
    list_del(&ops->list);
    plugin_num--;
    pthread_rwlock_unlock(&plugin_rwlock);
	return 0;
}

void *print_packet_counter(void * argv)
{
    long last_counter = 0;
    for(;;){
        printf("recv packet new %ld, total %ld file_send %ld color_send %ld traffic %ldMb/s\n",recv_counter-last_counter,recv_counter,send_counter,color_send_counter,(recv_traffic*8)/(5*1024*1024));
        recv_traffic = 0;
        last_counter = recv_counter;
        sleep(5);
    }
	return NULL;
}


void handler_signal(int signo)
{
    pc_plugin_ops *tmp_plugin_ops,*n_ops;

    printf("\n*********************************\nprepare to exit the main entry\n");
  
    printf("stop the counter thread\n");
    pthread_cancel(pid_counter);

    printf("stop the udp modules\n");
    finit_protocol_udp();
    
    printf("stop the tcp modules\n");
    finit_protocol_tcp();

    /*finish the policy modules*/
    printf("free the policy modules\n");
    finit_tlogd_policy();
    
    /*unload the plugin*/
    printf("prepare to unload all the plugs\n");
    pthread_rwlock_wrlock(&plugin_rwlock);
    list_for_each_entry_safe(tmp_plugin_ops,n_ops,&plugin_list,list)
    {
        list_del(&tmp_plugin_ops->list);
        tmp_plugin_ops->unload_plugin();
    }
    pthread_rwlock_unlock(&plugin_rwlock);
    pthread_rwlock_destroy(&plugin_rwlock);

    printf("free the memory pool\n");
    #ifdef _MEM_POOL
    mem_pool_destroy();
    #endif

    printf("success to exit the hplogd\n**********************************\n");
    exit(1);
}

/* pleae make sure, at least one plug is exist
 * infrastruct for the pkt processing
 */
inline int main_framer_process_pkt(char *msg)
{
    pc_plugin_ops *tmp_plugin_ops = NULL;
    pthread_rwlock_rdlock(&plugin_rwlock);
    list_for_each_entry(tmp_plugin_ops,&plugin_list,list)     
    {
        /*if the return value is PR_STOP, we stop to process the plugin*/
        if(PR_STOP == tmp_plugin_ops->process_log(msg)) 
        {
            break;
        }
        //continue to ask for the plugin                      
    }
    pthread_rwlock_unlock(&plugin_rwlock);
    return OK;
}

int main(int argc, char *argv[])
{
    //struct hostent *host;
    pc_plugin_ops *file_plugin_ops,*color_plugin_ops;
    pc_plugin_ops *tmp_plugin_ops,*n_ops;

    /*register the signal*/
    signal(SIGINT,handler_signal);

    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);

    pthread_mutex_init(&counter_mutex,NULL);
    pthread_create(&pid_counter,&attr,print_packet_counter,NULL);   

    /*initialize the plugin array*/
    INIT_LIST_HEAD(&plugin_list);
    pthread_rwlock_init(&plugin_rwlock,NULL);
   
    #ifdef _MEM_POOL_ 
    /*initialize the memory pool,no reason to fail*/
    if(FALSE == mem_pool_init(8192))//hard coded, each queue has 8196 free buffers Max.
    {
        perror("failed to initialize the memory pool \n");
        return FALSE;
    }
    #endif

    /*load and register the file plugin*/
    /*file handler is loaded staticly*/
    file_plugin_ops = file_initialize_plugin(MAX_LOG_SIZE,MAX_MAIN_QUEUE);
    if(!file_plugin_ops)
        return FALSE;
    /*if the num of plugin are too much, reject,but we don't care for the file plugin*/
    mc_register_plugin(file_plugin_ops);

    /*load and register the color plugin*/
    color_plugin_ops = color_initialize_plugin(MAX_LOG_SIZE,MAX_MAIN_QUEUE);
    if(!color_plugin_ops)
        goto UNREGISTER_FILE;
    /*if the num of plugin are too much, reject,but we don't care for the file plugin*/
    mc_register_plugin(color_plugin_ops);
   
    /*initialize the policy modules*/
    if(FALSE == init_tlogd_policy())
    {
        perror("failed to initialze the policy module");
        goto UNREGISTER_FILE;
    }

    /*initialize the udp protocol modules*/
    if(FALSE == init_protocol_udp())
    {
        goto FINIT_POLICY;
    }
   
    /*initialize the tcp protocol modules*/
    if(FALSE == init_protocol_tcp())
    {
        goto FINIT_UDP;
    }

    /*pending on the working thread*/
    pthread_join(pid_counter,NULL);
    protocol_udp_join();
    protocol_tcp_join();
    
    /*DONOT forget to free the buffer*/

    return OK;

FINIT_UDP:
    finit_protocol_udp();
FINIT_POLICY:
    finit_tlogd_policy();
UNREGISTER_FILE:
    /*unload the plugin*/
    pthread_rwlock_wrlock(&plugin_rwlock);
    list_for_each_entry_safe(tmp_plugin_ops,n_ops,&plugin_list,list)
    {
        list_del(&tmp_plugin_ops->list);
        tmp_plugin_ops->unload_plugin();
    }
    pthread_rwlock_unlock(&plugin_rwlock);
    /*destroy the lock*/
    pthread_rwlock_destroy(&plugin_rwlock);
    pthread_mutex_destroy(&counter_mutex);
    #ifdef _MEM_POOL_
    /*destroy the memory pool*/
    mem_pool_destroy();
    #endif
    return OK;
}

